/*
 * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */
package com.i360day.invoker;

import com.i360day.invoker.exception.HttpInvokerException;
import com.i360day.invoker.http.Request;
import com.i360day.invoker.http.Response;
import com.rabbitmq.client.*;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;

import java.io.ByteArrayInputStream;
import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author liju.z
 * @date 2024/3/26 21:10
 */
public class DefaultMessageReply extends DefaultConsumer implements Closeable {
    private final MessagePendingReply<byte[]> messagePendingReply = new MessagePendingReply<>();
    private String replyTo;
    private String consumerTag;
    private ShutdownListener shutdownListener;
    private Lock lock;

    /**
     * Constructs a new instance and records its association to the passed-in channel.
     *
     * @param channel the channel to which this consumer is attached
     */
    public DefaultMessageReply(String replyTo, Channel channel) throws IOException {
        super(channel);
        this.replyTo = replyTo;
        this.lock = new ReentrantLock();
        init();
    }

    public void setReplyTo(String replyTo) {
        this.replyTo = replyTo;
    }

    public String getReplyTo() {
        return replyTo;
    }

    @Override
    public String getConsumerTag() {
        return consumerTag;
    }

    public void setConsumerTag(String consumerTag) {
        this.consumerTag = consumerTag;
    }

    /**
     * 初始化
     *
     * @throws IOException
     */
    public void init() throws IOException {
        getChannel().queueDeclare(replyTo, true, true, false, null);
        getChannel().queueDeclarePassive(replyTo);
        getChannel().basicConsume(replyTo, true, this);
        //监听
        this.shutdownListener = c -> {
            if (!RabbitUtils.isNormalChannelClose(c)) {
                messagePendingReply.completeExceptionally(c);
            }
        };
        getChannel().addShutdownListener(shutdownListener);
    }

    /**
     * 监听返回数据包
     *
     * @param consumerTag
     * @param envelope
     * @param properties
     * @param body
     * @throws IOException
     */
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        setConsumerTag(consumerTag);
        messagePendingReply.reply(body);
    }

    @Override
    public void handleConsumeOk(String consumerTag) {
        setConsumerTag(consumerTag);
    }

    @Override
    public void handleCancelOk(String consumerTag) {
        setConsumerTag(consumerTag);
        messagePendingReply.returned(new HttpInvokerException("队列取消"));
    }

    @Override
    public void handleCancel(String consumerTag) throws IOException {
        setConsumerTag(consumerTag);
    }

    @Override
    public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
        setConsumerTag(consumerTag);
        messagePendingReply.completeExceptionally(sig);
    }

    @Override
    public void handleRecoverOk(String consumerTag) {
        setConsumerTag(consumerTag);
    }

    /**
     * 发送消息
     *
     * @param exchange
     * @param routing
     * @param properties
     * @param body
     * @throws IOException
     */
    public void sendMessage(String exchange, String routing, AMQP.BasicProperties properties, byte[] body) throws IOException {
        getChannel().basicPublish(exchange, routing, properties, body);
    }

    public Response getResponse(long timeout, TimeUnit unit, Request request) throws HttpInvokerException, IOException {
        try {
            //获取数据包
            byte[] bytes = messagePendingReply.get(timeout, unit);

            //返回信息
            return Response.Builder.create()
                    .body(new ByteArrayInputStream(bytes))
                    .status(bytes != null ? HttpStatus.OK.value() : HttpStatus.INTERNAL_SERVER_ERROR.value())
                    .setContextType(MediaType.APPLICATION_JSON_VALUE)
                    .request(request)
                    .build();

        } catch (Exception ex) {
            throw new HttpInvokerException(ex);
        } finally {
            try {
                getChannel().queueDelete(replyTo);
            } catch (Exception ex) {
                //ignore exception
            }
        }
    }

    private <T> void ignoreExceptionExecute(IgnoreException ignoreException){
        try{
            ignoreException.callback();
        }catch (Exception ex){

        }
    }

    public interface IgnoreException{
        void callback() throws Exception;
    }

    @Override
    public void close() {
        //删除销毁监听
        ignoreExceptionExecute(() -> getChannel().removeShutdownListener(shutdownListener));
        //删除客户端tag
        ignoreExceptionExecute(() -> getChannel().basicCancel(getConsumerTag()));
        //删除回复队列
        ignoreExceptionExecute(() -> getChannel().queueDelete(getReplyTo()));
    }
}
